package GraghX.algorithms

import org.apache.spark.graphx.{GraphLoader, VertexId, VertexRDD}
import org.apache.spark.{SparkConf, SparkContext}

object ConnectedComponentsDemo {
def main(args: Array[String]): Unit = {
  //设置运行环境
  val conf = new SparkConf().setAppName("ConnectedComponentsDemo").setMaster("local")
  val sc = new SparkContext(conf)
  sc.setLogLevel("warn")

  //导入数据
  val graph  = GraphLoader.edgeListFile(sc,"d:/tmp/followers.txt")

  //找图关联部分
  //创建RDD
  val cc:VertexRDD[VertexId] = graph.connectedComponents().vertices
  cc.collect.foreach(println(_))
  println("*********************************")

  //找用户的关联部分
  val users = sc.textFile("d:/tmp/users.txt").map{line =>
    val fields = line.split(",")
    (fields(0).toLong,fields(1))
  }

  val ccByUsername = users.join(cc).map {
    case (id , (username , cc ))=> (username,cc)
  }

  //结果输出
  print(ccByUsername.collect().mkString("\n"))
}
}
